package io.github.junxworks.qt.modules.mm.engine;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.util.List;
import java.util.Map;
import java.util.concurrent.SynchronousQueue;

import org.stringtemplate.v4.ST;

import com.google.common.collect.Maps;

import io.github.junxworks.ep.core.ds.DynamicDataSource;
import io.github.junxworks.junx.core.exception.FatalException;
import io.github.junxworks.junx.core.lifecycle.ThreadService;
import io.github.junxworks.junx.core.util.StringUtils;
import io.github.junxworks.qt.modules.mm.vo.FieldVo;
import io.github.junxworks.qt.modules.mm.vo.MmDatasourceVo;

/**
 * 数据流对象
 */
public class DatastreamStreamLogic extends ThreadService {

	private String dsName;

	private String sqlTemplate;

	private MmDatasourceVo datasource;

	private SynchronousQueue<DataContext> queue = new SynchronousQueue<>();

	private Map<String, StreamDataBundle> objCache = Maps.newConcurrentMap();

	private List<FieldVo> fields;

	private ExeContext ctx;

	private String sql;

	private DynamicDataSource dynamicDataSource;

	public List<FieldVo> getFields() {
		return fields;
	}

	public void setFields(List<FieldVo> fields) {
		this.fields = fields;
	}

	public DynamicDataSource getDynamicDataSource() {
		return dynamicDataSource;
	}

	public void setDynamicDataSource(DynamicDataSource dynamicDataSource) {
		this.dynamicDataSource = dynamicDataSource;
	}

	public Map<String, StreamDataBundle> getObjCache() {
		return objCache;
	}

	public String getDsName() {
		return dsName;
	}

	public void setDsName(String dsName) {
		this.dsName = dsName;
	}

	public ExeContext getCtx() {
		return ctx;
	}

	public void setCtx(ExeContext ctx) {
		this.ctx = ctx;
	}

	public void addJob() {
		queue.add(null);
	}

	public String getSqlTemplate() {
		return sqlTemplate;
	}

	public void setSqlTemplate(String sqlTemplate) {
		this.sqlTemplate = sqlTemplate;
	}

	public MmDatasourceVo getDatasource() {
		return datasource;
	}

	public void setDatasource(MmDatasourceVo datasource) {
		this.datasource = datasource;
	}

	@Override
	protected void onStart() throws Throwable {
		//初始化执行sql
		if (StringUtils.isNull(sqlTemplate)) {
			throw new FatalException("Null sql template.");
		}
		ST st = new ST(sqlTemplate, '{', '}');
		if (StringUtils.notNull(ctx.getTsCodesCondition())) {
			st.add("tsCodes", ctx.getTsCodesCondition());
		}
		st.add("tradeDate", ctx.getTradeDate());
		this.sql = st.render();
	}

	@Override
	protected void onStop() throws Throwable {

	}

	public void extractData(DataContext bundle) throws InterruptedException {
//		if (!isRunning() && !objCache.containsKey(bundle.getKey().getValue())) {
//			//数据流已经执行完了，没数据了，最后一个数据不是key对应数据
//			bundle.decreaseCount();
//		} else if (!isRunning() && objCache.containsKey(bundle.getKey().getValue())) {
//			//数据流已经执行完了，没数据了，最后一个数据便是key对应数据
//			bundle.addData(objCache.remove(bundle.getKey().getValue()));
//		} else {
//			this.queue.put(bundle);
//		}
	}

	@Override
	protected void onRun() throws Throwable {
		/*DataContext bundle = null;
		try {
			//			try (Connection conn = GenerateHelper.getConnection(datasource.getDbType(), datasource.getConnUrl(), datasource.getDbUsername(), datasource.getDbPasswd()); PreparedStatement ps = conn.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);) {
			try (Connection conn = dynamicDataSource.getConnection(); PreparedStatement ps = conn.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);) {
				ps.setFetchSize(Integer.MAX_VALUE);
				ResultSet rs = ps.executeQuery();
				ResultSetMetaData md = rs.getMetaData();
				int columnCount = md.getColumnCount();
				boolean continued = false;
				DataObject targetData = null;
				while (rs.next()) {
					if (!continued) {
						bundle = queue.take(); //阻塞等待任务
					}
					Key k = bundle.getKey();
					String key = k.getValue();
					if (objCache.containsKey(key)) {
						//防止切换key的时候漏数据
						targetData = objCache.get(key);
					} else {
						targetData = new DataObject();
						targetData.setKey(key);
						targetData.setDataStreamName(dsName);
						objCache.put(key, targetData);
					}
					Map<String, Object> row = Maps.newHashMap();
					for (int i = 1; i <= columnCount; i++) {
						String columnName = md.getColumnName(i);
						row.put(columnName, rs.getObject(i));
					}
					String rowKey = String.valueOf(row.get(k.getColumn()));//数据库中的key
					if (k.getValue().equals(rowKey)) {
						//分支一，是要找的数据
						targetData.addData(row);
						continued = true;
					} else {
						//分支二，切换key了， 把上个数据加入到bundle
						targetData = new DataObject();
						targetData.setKey(rowKey);
						targetData.setDataStreamName(dsName);
						targetData.addData(row);
						objCache.put(rowKey, targetData);
		
						if (objCache.containsKey(key)) {
							bundle.addData(objCache.remove(key));
						} else {
							//防止没有找到股票数据
							bundle.decreaseCount();
						}
						continued = false;
					}
				}
				if (continued) {
					//如果最后有2条以上同一个stock数据，则会总分支1结束
					if (objCache.containsKey(bundle.getKey().getValue())) {
						bundle.addData(objCache.remove(bundle.getKey().getValue()));
					}
				}
			}
		} catch (Exception e) {
			if (bundle != null) {
				bundle.decreaseCount();
			}
			throw e;
		} finally {
			this.stop();
		}*/
	}

}
